package lucene.server.cdc.mysql;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.extern.slf4j.Slf4j;
import lucene.server.cdc.client.AbstractClient;
import lucene.server.cdc.util.SqlParser;
import lucene.server.cdc.mysql.model.BinlogPosition;
import lucene.server.cdc.mysql.util.BinlogUtil;
import lucene.server.cdc.mysql.util.ConvertUtil;
import lucene.server.common.enums.TaskEnum;
import lucene.server.common.event.EventData;
import lucene.server.common.storage.StorageHelper;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
public class MysqlBinlogClient extends AbstractClient {

    private BinaryLogClient                     client;
    private Timer                               storageTimer;
    private boolean                             isDealEvent = true;
    private BinlogPosition                      binlogPosition;
    private ConcurrentHashMap<Long,  String >   tableMap;

    public MysqlBinlogClient(Properties properties) {
        super(properties);
        this.loadStorage();
        client = configureBinaryLogClient();
        setClientBinlogPosition();
    }

    private BinaryLogClient configureBinaryLogClient() {
        client = new BinaryLogClient(
                properties.getProperty("host"),
                Integer.parseInt(properties.getProperty("port")),
                properties.getProperty("username"),
                properties.getProperty("password")
        );
        client.setHeartbeatInterval(10 * 1000L);

        // 设置反序列化配置
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
         );
        client.setEventDeserializer(eventDeserializer);
        client.setKeepAlive(true);
        client.setKeepAliveInterval(10 * 1000L);
        client.registerEventListener(this::handleEvent);
        client.registerLifecycleListener(new ClientLifeCycleListener(properties));
        client.setServerId(getRandomServerId());
        return client;
    }
    private void handleEvent(Event event) {
        EventType eventType = event.getHeader().getEventType();
        if (EventType.isRowMutation(eventType)) {
           processRowMutationEvent(event);
        } else if (eventType == EventType.TABLE_MAP) {
           processTableMapEvent(event.getData());
        }else if (eventType == EventType.QUERY) {
            processQueryEvent(event.getData());
        }
        updateBinlogPosition(eventType);
    }
    private void processQueryEvent(QueryEventData queryEventData){

        String database = queryEventData.getDatabase();
        if (!this.filter.containsDatabase(database)){
            return;
        }
        String sql = queryEventData.getSql();
        //如果不是表操作相关的
        if(SqlParser.determineDDLType(sql)==null){
            return;
        }
        //todo 这里的准确性还需要确认
        String tableName= SqlParser.extractTableName(sql);
        if(tableName==null){
            return;
        }
        if (this.filter.containsTable(database,tableName)){
            //需要重新加载表的结构
            System.out.println("需要重新加载表的结构"+database+"::"+tableName);
            filter.refreshColumn(database,tableName);
        }

    }
    private void processTableMapEvent(TableMapEventData tableMapEventData) {
        String database = tableMapEventData.getDatabase();
        String table = tableMapEventData.getTable();
        long tableId = tableMapEventData.getTableId();
        if (this.filter.containsTable(database , table)) {
            this.tableMap.put(tableId, database + "::" + table);
        }
    }
    private void processRowMutationEvent(Event event) {
            EventType eventType = event.getHeader().getEventType();
            List<EventData> changeDatas = new ArrayList<>();
            if (EventType.isUpdate(eventType)) {
                UpdateRowsEventData data = event.getData();
                if (!tableMap.containsKey(data.getTableId())) {
                    return;
                }
                String tableInfo = tableMap.get(data.getTableId());
                data.getRows().forEach(mapEntry -> {
                    EventData changeData = new EventData(TaskEnum.UPDATE,tableInfo);
                    ConvertUtil.getUpdateMap(changeData,mapEntry, this.filter.getTableColumns(tableInfo));
                    changeDatas.add(changeData);
                });
            } else if (EventType.isWrite(eventType)) {
                WriteRowsEventData data = event.getData();
                if (!tableMap.containsKey(data.getTableId())) {
                    return;
                }
                String tableInfo = tableMap.get(data.getTableId());
                data.getRows().forEach(mapEntry -> {
                    EventData changeData = new EventData(TaskEnum.INSERT,tableInfo);
                    ConvertUtil.getInsertOrDeleteMap(changeData,mapEntry, this.filter.getTableColumns(tableInfo));
                    changeDatas.add(changeData);
                });
            } else if (EventType.isDelete(eventType)) {
                DeleteRowsEventData data = event.getData();
                if (!tableMap.containsKey(data.getTableId())) {
                    return;
                }
                String tableInfo = tableMap.get(data.getTableId());
                data.getRows().forEach(mapEntry -> {
                    EventData changeData = new EventData(TaskEnum.DELETE,tableInfo);
                    ConvertUtil.getInsertOrDeleteMap(changeData,mapEntry, this.filter.getTableColumns(tableInfo));
                    changeDatas.add(changeData);
                });
            }
            process(changeDatas);
    }

    private void updateBinlogPosition(EventType event) {

        if(EventType.HEARTBEAT.equals(event)){
            return;
        }
        //如果大于0，说明当前binlog文件名比当前存储的binlog文件名大
        int nameCompare = client.getBinlogFilename().compareTo(binlogPosition.getBinlogName());

        //这里是当前的binlog文件名比当前存储的binlog文件名大
        if(nameCompare==0) {
            //这里是当前的binlog文件名比当前存储的binlog文件名相同
            //但是当前binlog的position比当前存储的position大
            if (client.getBinlogPosition()>binlogPosition.getPosition()){
                binlogPosition.setPosition(client.getBinlogPosition());
                isDealEvent = true;
            }else{
                isDealEvent = false;
            }
        } else if(nameCompare>0){
            binlogPosition.setPosition(client.getBinlogPosition());
            binlogPosition.setBinlogName(client.getBinlogFilename());
            isDealEvent = true;
        } else{
            //这里是当前的binlog文件名比当前存储的binlog文件名小，不需要处理这些数据
            isDealEvent = false;
        }
    }
    //设置binlog位置
    private void setClientBinlogPosition()  {
        String binlogFilePath =  this.name+"_bin.cdc";
        BinlogPosition iniBinlogPosition;
        try {
            binlogPosition = (BinlogPosition) StorageHelper.fromFile(binlogFilePath);
            iniBinlogPosition = BinlogUtil.getIniBinlogPosition(properties, binlogPosition);
            if (binlogPosition ==null){
                binlogPosition = new BinlogPosition(iniBinlogPosition.getBinlogName(),iniBinlogPosition.getPosition());
            }
            client.setBinlogPosition(iniBinlogPosition.getPosition());
            client.setBinlogFilename(iniBinlogPosition.getBinlogName());
        }catch (Exception e) {
            e.printStackTrace();
        }

    }


    private void loadStorage() {

       String tableMapPath = this.name+"_table.cdc";
       //加载 table id对照表
       this.tableMap = (ConcurrentHashMap<Long,  String >) StorageHelper.fromFile(tableMapPath);
       if (this.tableMap == null){
          this.tableMap = new  ConcurrentHashMap<>();
       }

    }

    private void storageSchedule() {
        storageTimer = new Timer();
        storageTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                saveBinlogPosition();
            }
        }, 1, 2000);
    }
    private void saveBinlogPosition() {

        StorageHelper.toFile(binlogPosition,  this.name+"_bin.cdc");
        StorageHelper.toFile(tableMap,  this.name+"_table.cdc");

    }

    @Override
    public void start() {
        new Thread(() -> {
            try {
                client.connect();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
        storageSchedule();
    }

    @Override
    public void reStart() {
        try {
            client.setBinlogFilename(null);
            client.setBinlogPosition(4L);
            client.connect();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void shutdown(){
        try {
            storageTimer.cancel();
            client.disconnect();
            saveBinlogPosition();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean isRunning() {
        return client.isConnected();
    }

    long getRandomServerId() {
        Random random = new Random();
        return random.nextLong();
    }
}
